package com.demo.zookeeper.zkmanager;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.util.Collection;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName ZkClient
 * @Author luoguoqiang
 * @Description TODO
 * @date 2021/3/19 14:51
 * @Version 1.0
 */
@Component
@Slf4j
public class ZkClient {

  @Value("${zookeeper.address.list}")
  private String zkConnectString;

  private CuratorFramework curatorFramework;

  public CuratorFramework connectZk(){
    CuratorFramework client = CuratorFrameworkFactory.builder()
        .connectString(zkConnectString)
        .sessionTimeoutMs(5000)
        .connectionTimeoutMs(5000)
        .retryPolicy(new ExponentialBackoffRetry(1000, 3))
        .namespace("curator")
        .build();
    this.curatorFramework = client;
    client.start();
    log.info("{} zookeeper start. root is /curator", zkConnectString);
    return client;
  }

  public boolean addNode(String node, String dataValue){
    try {
      curatorFramework.create()
          .creatingParentsIfNeeded()
          .forPath(node, dataValue.getBytes());
      log.info("create {} node success", node);
      return true;
    } catch (Exception e) {
      log.error("{}",e );
      return false;
    }
  }

  public boolean deleteNote( String node){
    try {
      curatorFramework.delete()
          .deletingChildrenIfNeeded()
          .forPath(node);
      log.info("delete {} node success", node);
      return true;
    } catch (Exception e) {
      log.error("{}",e );
      return false;
    }

  }

  public byte[] getNodeData(String node, Stat stat){
    try {
      return curatorFramework.getData()
          .storingStatIn(stat)
          .forPath(node);
    } catch (Exception e) {
      log.error("{}",e );
      return null;
    }
  }

  public Stat modifyNodeData(String node, byte[]  value){
    try {
      return curatorFramework.setData().forPath(node, value);
    } catch (Exception e) {
      log.info("{}", e);
      return null;
    }
  }

  public void closeZk(){
    CloseableUtils.closeQuietly(curatorFramework);
  }

  public void zkTransactionRollback(){
    try {
      log.info("开始zookeeper事务操作");
      Collection<CuratorTransactionResult> results = curatorFramework.inTransaction()
          .create().forPath("/t1", "111".getBytes())
          // 操作一个正确节点
          .and().setData().forPath("/t2", "111".getBytes())
          // 操作一个不存在的节点 会报错并回退
          .and().setData().forPath("/xxxx", "111".getBytes())
          .and().commit();
      for (CuratorTransactionResult c : results) {
        log.info("{}", c);
      }
    } catch (Exception e) {
      log.info("{}", e);
    }

  }

  public void zkTransactionCommit(){
    try {
      log.info("开始zookeeper事务操作");
      Collection<CuratorTransactionResult> results = curatorFramework.inTransaction()
          .create().withMode(CreateMode.EPHEMERAL).forPath("/t1", "111".getBytes())
          // 操作一个正确节点
          .and().setData().forPath("/t2", "111".getBytes())
          .and().commit();
      for (CuratorTransactionResult c : results) {
        log.info("{}", c);
      }
    } catch (Exception e) {
      log.info("{}", e);
    }
  }

  public void zkListen() throws Exception {
    PathChildrenCache cache = new PathChildrenCache(curatorFramework, "/t2", true);

    cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
    cache.getListenable().addListener((curatorFramework1, pathChildrenCacheEvent) -> {
      // 事件还有很多,列举三个
      switch (pathChildrenCacheEvent.getType()) {
        case CHILD_ADDED:
          log.info("增加子节点 : {}", pathChildrenCacheEvent.getData());
          break;
        case CHILD_REMOVED:
          log.info("删除子节点 : {}", pathChildrenCacheEvent.getData());
          break;
        case CHILD_UPDATED:
          log.info("更新子节点 : {}", pathChildrenCacheEvent.getData());
          break;
        default:
          break;
      }
    });
    curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath("/t2/e1","e1".getBytes());
    TimeUnit.SECONDS.sleep(1);

    curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath("/t2/e2","e2".getBytes());
    TimeUnit.SECONDS.sleep(1);

    curatorFramework.setData().forPath("/t2/e1", "222".getBytes());
    TimeUnit.SECONDS.sleep(1);
    curatorFramework.setData().forPath("/t2/e1", "333".getBytes());
    TimeUnit.SECONDS.sleep(1);

    curatorFramework.delete().forPath("/t2/e2");

    CloseableUtils.closeQuietly(cache);
  }


}
